package com;

import com.alibaba.fastjson.JSONObject;
import com.process.SqlParse;
import com.dto.User;
import com.jdbc.SpringJDBC;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.time.LocalDateTime;

/**
 * @author 惊云
 * @date 2023/8/13 15:58
 */
public class CustomSink extends RichSinkFunction<String> {

    @Override
    public void invoke(String value, Context context) throws Exception {

        System.out.println("监听到活动数据:" + LocalDateTime.now() + value);

        JSONObject jsonObject = JSONObject.parseObject(value);

        User before = jsonObject.getObject("before", User.class);
        User after = jsonObject.getObject("after", User.class);

        try {
            String table = jsonObject.getJSONObject("source").getString("table");

            SqlParse sqlParse = new SqlParse();
            String executeSQL = "";
            if(before == null){
                // 插入
                executeSQL = sqlParse.getInsert(after,table);
            }else if(after == null){
                // 删除
                executeSQL = sqlParse.getDeleteSQL(before,table);
            }
            else{
                // 更新
                executeSQL = sqlParse.getUpdateSQL(before,after,table);
            }
            SpringJDBC.executeSQL(executeSQL);
        }catch (Exception e){
            System.out.println("执行错误");
        }

    }

}
